package com.atguigu.hightest;

import com.atguigu.bean.UserBehavior;
import com.atguigu.utils.ITtoListUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 需求：指定时间范围内网站独立访客数（UV）的统计
 * 需要对pv操作按照用户id进行一个去重
 */

public class Flink03_UV {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(2);

        DataStreamSource<String> ds = env.readTextFile("input/UserBehavior.csv");

        ds.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String value) throws Exception {
                String[] datas = value.split(",");
                return new UserBehavior(
                        Long.valueOf(datas[0]),
                        Long.valueOf(datas[1]),
                        Integer.valueOf(datas[2]),
                        datas[3],
                        Long.valueOf(datas[4]) * 1000
                );
            }
        })
                //加水印
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        //设置最大乱序时间
                        .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //设置和水印的比较值
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        })
                )
                //筛选元素
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior value) throws Exception {
                        return "pv".equals(value.getBehavior());
                    }
                })
                //keyby
                .keyBy(UserBehavior::getBehavior)
                //开窗
                .window(TumblingEventTimeWindows.of(Time.minutes(30)))
                //处理
                .process(new ProcessWindowFunction<UserBehavior, String, String, TimeWindow>() {
                    private MapState<Long, Integer> mapState;

                    //需要对pv的操作按照用户id进行一个去重
                    //设置一个map类型的状态

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("mapState", Long.class, Integer.class));
                    }

                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<UserBehavior> elements,
                                        Collector<String> out) throws Exception {
                        //时刻要清楚的知道键控状态只和key有关
                        mapState.clear();

                        for (UserBehavior element : elements) {
                            mapState.put(element.getUserId(),1);
                        }

                        //结果输出
                        out.collect(context.window()+" "+ ITtoListUtil.toList(mapState.keys()).size());

                    }
                })
                .print();





        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
